package com.seal.commons.infrastructure.rocketmq.autoConfigure;

import com.seal.commons.infrastructure.facade.jms.MessageListener;
import com.seal.commons.infrastructure.facade.jms.MessageSender;
import com.seal.commons.infrastructure.rocketmq.RocketMQListenerAutoExporter;
import com.seal.commons.infrastructure.rocketmq.listener.DefaultMessageListener;
import com.seal.commons.infrastructure.rocketmq.sender.RocketMQSender;
import com.seal.commons.utils.NetUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.EnableMBeanExport;
import org.springframework.context.annotation.Import;
import org.springframework.jmx.support.RegistrationPolicy;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ClassName:RocketMQConfxiguration <br/>
 * Date:     2019-05-31 13:41 <br/>
 *
 * @author yong.huang
 * @see
 * @since JDK 1.8
 */
@Configuration
@ConditionalOnClass({RocketMQProperties.class})
@Import(RocketMQProperties.class)
@EnableMBeanExport(registration = RegistrationPolicy.IGNORE_EXISTING)
@Slf4j
public class RocketMQConfiguration {

    @Value("${spring.application.name}")
    private String application;

    @Autowired
    private RocketMQProperties rocketMQProperties;

    @Bean(value = "rocketMqProducer", initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQProducer getDefaultMQProducer() {
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
        defaultMQProducer.setProducerGroup(application + "-producer");
        defaultMQProducer.setNamesrvAddr(rocketMQProperties.getNameSrvAdder());
        RocketMQProperties.Producer producer = rocketMQProperties.getProducer();
        defaultMQProducer.setDefaultTopicQueueNums(producer.getDefaultTopicQueueNums());
        defaultMQProducer.setSendMsgTimeout(producer.getSendMsgTimeout());
        defaultMQProducer.setMaxMessageSize(producer.getMaxMessageSize());
        defaultMQProducer.setCompressMsgBodyOverHowmuch(producer.getCompressMsgBodyOverHowmuch());
        defaultMQProducer.setRetryTimesWhenSendFailed(producer.getRetryTimesWhenSendFailed());
        defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(producer.isRetryAnotherBrokerWhenNotStoreOK());
        defaultMQProducer.setClientIP(NetUtils.getLocalHost());
        return defaultMQProducer;
    }


    @Bean("rocketMQSender")
    public MessageSender rocketMQSender(){
        return new RocketMQSender(getDefaultMQProducer());
    }

    @Bean("listenerMap")
    public Map<String, MessageListener> getListenerMap(){
        return new ConcurrentHashMap<>();
    }


    @Bean
    public DefaultMessageListener getDefaultMessageListener(){
        return new DefaultMessageListener(getListenerMap());
    }

    /**
     *  广播模式
     */
    @Bean(value = "broadcastingMQPushConsumer",initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQPushConsumer broadcastingMQPushConsumer(){
        return build(application+"-broadcasting-consumer", rocketMQProperties.getNameSrvAdder(),
                rocketMQProperties.getConsumer(), MessageModel.BROADCASTING);
    }

    /**
     * 集群模式
     */
    @Bean(value = "clusteringMQPushConsumer",initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQPushConsumer clusteringMQPushConsumer(){
        return build(application+"-clustering-consumer", rocketMQProperties.getNameSrvAdder(),
                rocketMQProperties.getConsumer(), MessageModel.CLUSTERING);
    }

    @Bean
    @ConditionalOnMissingBean
    public RocketMQListenerAutoExporter rocketMQListenerAutoExporter(){
        RocketMQListenerAutoExporter exporter =
                new RocketMQListenerAutoExporter(getListenerMap(), broadcastingMQPushConsumer(), clusteringMQPushConsumer());
        return exporter;
    }

    private DefaultMQPushConsumer build(String groupName, String namesrvAddr, RocketMQProperties.Consumer consumer, MessageModel messageModel){
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(groupName);
        defaultMQPushConsumer.setNamesrvAddr(namesrvAddr);
        defaultMQPushConsumer.setMessageModel(messageModel);
        defaultMQPushConsumer.setConsumeThreadMin(consumer.getConsumeThreadMin());
        defaultMQPushConsumer.setConsumeThreadMax(consumer.getConsumeThreadMax());
        defaultMQPushConsumer.setConsumeConcurrentlyMaxSpan(consumer.getConsumeConcurrentlyMaxSpan());
        defaultMQPushConsumer.setMessageListener(getDefaultMessageListener());
        defaultMQPushConsumer.setPullThresholdForQueue(consumer.getPullThresholdForQueue());
        defaultMQPushConsumer.setPullInterval(consumer.getPullInterval());
        defaultMQPushConsumer.setPullBatchSize(consumer.getPullBatchSize());
        return defaultMQPushConsumer;
    }

}
